/* Copyright (c) 2008, dZettaApart.
   Licensed under the BSD License.
   See COPYING for details. */

#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>

#include <cstdlib>
#include <deque>
#include <iostream>

#include "memcacheasio_connectionspool.h"
#include "memcacheasio_connection.h"

namespace memcacheasio
{
	/** ***************************************************************************
	 *  ConnectionsPool::ConnectionsPool
	 *
	 *****************************************************************************/
	ConnectionsPool::ConnectionsPool ( boost::asio::io_service& io_service ) :
			_host ( "" ),
			_port ( "" ),
			_io_service ( io_service ),
			_try_timer ( io_service ),
			_try_timer_connect ( io_service )
	{
		_try_count = 0;
		_limitConnection = 10;
		_try_connect = 0;
		_timeout_request = 5;
		_timeout_connect = 2;
		_stopped = false;
		_countConnection = 0;
	}

	/** ***************************************************************************
	 *  ConnectionsPool::~ConnectionsPool
	 *
	 *****************************************************************************/
	ConnectionsPool::~ConnectionsPool()
	{
	}

	/** **************************************************************************
	 * ConnectionsPool::Init
	 * нельзя в конструкторе делать  shared_from_this()
	 ****************************************************************************/
	void ConnectionsPool::init ( const std::string& host, const std::string& port, int count_connection, int tryCount, int tryTimeout, unsigned int limitConnection )
	{
        std::cout << "Connect to host: " << host << ":" << port << std::endl;
		_host = host;
		_port = port;
		_try_count = tryCount;
//		_try_timeout = tryTimeout;
		_limitConnection = limitConnection;

		for ( int i =0; i<count_connection; i++ )
		{
			newConnection ();
		}
	}


	/** ***************************************************************************
	 *  ConnectionsPool::stop
	 *
	 *****************************************************************************/
	void ConnectionsPool::stop()
	{
		boost::mutex::scoped_lock lock ( _mutex );

		_stopped = true;

		//Close all connection
		while ( !_available_connections.empty() )
		{
			Connection::Ptr c_connection = _available_connections.front();
			c_connection->close();
			_available_connections.pop_front();
		}

		while ( !_serve_connections.empty() )
		{
			Connection::Ptr c_connection = _serve_connections.front();
			c_connection->close();
			_serve_connections.pop_front();
		}

		while ( !_new_connections.empty() )
		{
			Connection::Ptr c_connection = _new_connections.front();
			c_connection->close();
			_new_connections.pop_front();
		}

		_try_timer.cancel();
		_try_timer_connect.cancel();

		while ( !_request_queue.empty() )
		{
			Connection::Ptr c_connection;
			cb_addConnection cb =  _request_queue.front();
			_request_queue.pop_front();
			boost::system::error_code error ( boost::asio::error::not_connected );
			cb ( c_connection, error );
		}
	}

	/** ***************************************************************************
	 *  ConnectionsPool::newConnection
	 *
	 *	Start async connect
	 *****************************************************************************/
	void ConnectionsPool::newConnection ( )
	{
		if ( _stopped )
			return;

		Connection::Ptr connection ( new Connection ( _io_service ) );
		connection->connect ( _host, _port, boost::bind ( &ConnectionsPool::queueConnection, shared_from_this(), _1, connection ) );
		_new_connections.push_back ( connection );


		boost::mutex::scoped_lock lock ( _mutex );
		_countConnection++;
	}

	/** ***************************************************************************
	 *  ConnectionsPool::getConnection
	 *
	 *	Get connection from connection pool
	 *	Async operation
	 *****************************************************************************/
	void ConnectionsPool::getConnection ( cb_addConnection cb/*, int timeout*/ )
	{
		boost::mutex::scoped_lock lock ( _mutex );

		if ( !_available_connections.empty() )
		{
			Connection::Ptr c_connection = _available_connections.front();
			_available_connections.pop_front();
			_serve_connections.push_back ( c_connection );
			boost::system::error_code error;
			_io_service.post ( boost::bind ( cb, c_connection, error ) );
		}
		else
		{
			_request_queue.push_back ( cb );

			int res = _try_timer.expires_from_now ( boost::posix_time::seconds ( _timeout_request ) );
			if ( !res )
			{
				_try_timer.async_wait ( boost::bind ( &ConnectionsPool::checkAvailableConnection, shared_from_this(),
				                                      boost::asio::placeholders::error ) );
			}
		}
	}

	/** ***************************************************************************
	 *  ConnectionsPool::queueConnection
	 *
	 *****************************************************************************/
	void ConnectionsPool::queueConnection ( const boost::system::error_code& err, Connection::Ptr connection )
	{
		boost::mutex::scoped_lock lock ( _mutex );
		if ( connection )
		{
			_serve_connections.remove ( connection );
			_new_connections.remove ( connection );

			if ( _stopped )
				return;

			//Check connection
			if ( connection->is_open() && !err )
			{
				_available_connections.push_back ( connection );
			}
			else
			{
				if ( ( _available_connections.size() +
				        _serve_connections.size() +
				        _new_connections.size() ) < _limitConnection )
				{
					_try_connect++;
					if ( _try_connect < 3 )
					{
						newConnection();
					}
					else
					{
						int res = _try_timer_connect.expires_from_now ( boost::posix_time::seconds ( _timeout_connect ) );
						if ( !res )
						{
							_try_timer_connect.async_wait ( boost::bind ( &ConnectionsPool::timeoutNewConnection, shared_from_this(),
							                                boost::asio::placeholders::error ) );
						}
					}
				}
			}
		}

		if ( !_available_connections.empty() && !_request_queue.empty() )
		{
			Connection::Ptr c_connection = _available_connections.front();
			_available_connections.pop_front();
			_serve_connections.push_back ( c_connection );
			cb_addConnection cb = _request_queue.front();
			_request_queue.pop_front();
			boost::system::error_code error;
			_io_service.post ( boost::bind ( cb, c_connection, error ) );
		}

	}

	/** ***************************************************************************
	 *  ConnectionsPool::checkAvailableConnection
	 *
	 *****************************************************************************/
	void ConnectionsPool::checkAvailableConnection ( const boost::system::error_code& err )
	{
		bool aborted = ( err == boost::asio::error::operation_aborted ) ? true : false;
		if ( aborted )
			return;

		boost::mutex::scoped_lock lock ( _mutex );

		Connection::Ptr c_connection;
		boost::system::error_code error;

		if ( !_request_queue.empty() )
		{
			cb_addConnection cb = _request_queue.front();
			_request_queue.pop_front();

			if ( !_available_connections.empty() )
			{
				c_connection = _available_connections.front();
				_available_connections.pop_front();
				_serve_connections.push_back ( c_connection );
			}
			else
			{
				error = boost::asio::error::not_connected;
			}
			_io_service.post ( boost::bind ( cb, c_connection, error ) );
		}
	}

	/** ***************************************************************************
	 *  ConnectionsPool::timeoutNewConnection
	 *
	 *****************************************************************************/
	void ConnectionsPool::timeoutNewConnection ( const boost::system::error_code& err )
	{
		bool aborted = ( err == boost::asio::error::operation_aborted ) ? true : false;
		if ( aborted )
			return;

		boost::mutex::scoped_lock lock ( _mutex );

		_try_connect = 0;

		newConnection();
	}

} //namespace memcacheasio


